package com.huan.kafka.api;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 * 自定义生产者消息拦截器，此类不是线程安全的，需要自己控制线程安全问题
 *
 * @author huan.fu 2021/1/5 - 上午10:04
 */
public class CustomProducerInterceptor implements ProducerInterceptor<String, String> {

    @Override
    public void configure(Map<String, ?> configs) {

    }

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        System.out.println("此方法是在消息发送的业务线程中,可以在消息发送到kafka服务器时做一些处理");
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        System.out.println("此方法是在kafka 生产者的 I/O 线程中执行，需要立即返回，否则影响消息发送的速度");
        if (exception != null) {
            System.out.println("消息发送到服务器时发生了异常");
            return;
        }
        System.out.println("消息发送到了服务器 - offset:" + metadata.offset() + " partition:" + metadata.partition());
    }

    @Override
    public void close() {
        System.out.println("释放资源");
    }
}
